package ar.edu.itba.pod.legajo45129.balance.api;

import java.rmi.NotBoundException;
import java.rmi.RemoteException;
import java.rmi.registry.LocateRegistry;
import java.rmi.registry.Registry;
import java.rmi.server.UnicastRemoteObject;
import java.util.List;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicLong;

import org.apache.log4j.Logger;

import ar.edu.itba.balance.api.AgentsBalancer;
import ar.edu.itba.balance.api.NodeAgent;
import ar.edu.itba.balance.api.NotCoordinatorException;
import ar.edu.itba.node.Node;
import ar.edu.itba.node.NodeInformation;
import ar.edu.itba.pod.legajo45129.manager.NodeManager;

public class AgentsBalancerImpl implements AgentsBalancer {

	private static Logger logger = Logger.getLogger(AgentsBalancerImpl.class);

	private AtomicBoolean stopElection = new AtomicBoolean(false);
	private volatile NodeInformation coordinator;
	private AtomicLong timestampCoordinator = new AtomicLong();
	// private NodeInformation bullyElectionRequestNode;
	private AtomicLong timestampBullyElection = new AtomicLong();
	// private NodeInformation localNode;
	private ConcurrentMap<NodeInformation, Long> bullyElectionRequestNodes;

	private Thread coordinatorThread;

	protected static final long WAIT_FOR_BULLY_OK = 2000;

	public AgentsBalancerImpl() throws RemoteException {
		UnicastRemoteObject.exportObject(this, 0);
		this.bullyElectionRequestNodes = new ConcurrentHashMap<NodeInformation, Long>();
		NodeInformation localNode = NodeManager.getInstance()
				.getNodeInformation();
		Registry registry;
		registry = LocateRegistry.getRegistry(localNode.host(),
				localNode.port());
		if (registry == null) {
			registry = LocateRegistry.createRegistry(localNode.port());
		}

		logger.debug("Binding " + Node.AGENTS_BALANCER);
		registry.rebind(Node.AGENTS_BALANCER, this);
	}

	@Override
	public void bullyElection(NodeInformation node, long timestamp)
			throws RemoteException {
		// if (this.stopElection.get() == true) {
		// return;
		// }
		
		

		if (bullyElectionRequestNodes.containsKey(node)
				&& bullyElectionRequestNodes.get(node) == timestamp) {
			logger.debug("bullyElection message repeated");
			return;
		}

		if (this.coordinatorThread != null) {
			logger.debug("interrupt " + this.coordinatorThread.getName());
			this.coordinatorThread.interrupt();
		}

		bullyElectionRequestNodes.put(node, timestamp);
		logger.debug("bullyElection from node: " + node);
		this.stopElection.set(false);

		if (NodeManager.getInstance().isGreaterThan(node.id())
				&& this.timestampBullyElection.get() < timestamp) {

			NodeInformation localNode = NodeManager.getInstance()
					.getNodeInformation();

			// bullyOk and broadcast
			try {
				if (!localNode.equals(node)) {
					NodeManager.getInstance()
							.getAgentsBalancer(node.host(), node.port())
							.bullyOk(localNode);
				}

				this.timestampBullyElection.set(System.currentTimeMillis());
				final Thread t = broadcastBullyElection(localNode,
						this.timestampBullyElection.get());
				t.start();
				this.coordinatorThread = new Thread() {
					@Override
					public void run() {
						try {
							logger.debug(this.getName()
									+ " waiting for bullyOk");
							sleep(WAIT_FOR_BULLY_OK);
							if (stopElection.get() == false) {
								// if there is no bullyOk message received, then
								// I
								// set my self as coordinator;
								coordinator = NodeManager.getInstance()
										.getNodeInformation();
								synchronized (coordinator) {
									NodeManager.getInstance().setCoordinator(
											coordinator);
									broadcastCoordinator(coordinator,
											System.currentTimeMillis());
								}
							}
						} catch (InterruptedException e) {
							return;
						}
					}

				};
				this.coordinatorThread.start();

			} catch (NotBoundException e) {
				logger.warn(e);
			}
		} else {
			// forward messages
			if (!this.bullyElectionRequestNodes.containsKey(node)
					|| this.bullyElectionRequestNodes.get(node) != timestamp) {
				logger.debug("bullyElection forward");
				try {
					Thread t = broadcastBullyElection(node, timestamp);
					t.start();

				} catch (NotBoundException e) {
					logger.warn(e);
				}
			}
		}
	}

	private void broadcastCoordinator(NodeInformation node, long timestamp) {
		try {
			for (NodeInformation n : NodeManager.getInstance()
					.getClusterAdministration().connectedNodes()) {
				if (!stopElection.get()) {
					logger.debug("broadcast bullyCoordinator from " + node
							+ " to " + n);
					NodeManager.getInstance()
							.getAgentsBalancer(n.host(), n.port())
							.bullyCoordinator(node, timestamp);
				} else {
					logger.debug("election stopped");
				}
			}
		} catch (RemoteException e) {
			logger.warn(e);
		} catch (NotBoundException e) {
			logger.warn(e);
		}
	}

	private Thread broadcastBullyElection(final NodeInformation node,
			final long timestamp) throws RemoteException, NotBoundException {
		return new Thread() {
			@Override
			public void run() {
				NodeInformation localNode = NodeManager.getInstance()
						.getNodeInformation();
				try {
					for (NodeInformation n : NodeManager.getInstance()
							.getClusterAdministration().connectedNodes()) {
						if (!stopElection.get()) {
							logger.debug("broadcast bullyElection from "
									+ localNode + " to " + n);
							NodeManager.getInstance()
									.getAgentsBalancer(n.host(), n.port())
									.bullyElection(node, timestamp);
						} else {
							logger.debug("election stopped");
						}
					}
				} catch (RemoteException e) {
					logger.warn(e);
				} catch (NotBoundException e) {
					logger.warn(e);
				}
			}
		};
	}

	@Override
	public void bullyOk(NodeInformation node) throws RemoteException {
		logger.info("bullyOk from: " + node);
		this.stopElection.set(true);
	}

	@Override
	public void bullyCoordinator(NodeInformation node, long timestamp)
			throws RemoteException {

		if (this.coordinator == null
				|| this.timestampCoordinator.get() < timestamp) {
			logger.info("bullyCoordinator: " + node);
			this.coordinator = node;
			this.timestampCoordinator.set(timestamp);
			NodeManager.getInstance().setCoordinator(node);

			for (NodeInformation n : NodeManager.getInstance()
					.getClusterAdministration().connectedNodes()) {
				try {
					NodeManager.getInstance()
							.getAgentsBalancer(n.host(), n.port())
							.bullyCoordinator(node, timestamp);
				} catch (NotBoundException e) {
					logger.error(e.getMessage());
				}
			}
		} else {
			logger.debug("bullyCoordinator message repeated"
					+ " | timestampCoord < timestamp ->"
					+ (this.timestampCoordinator.get() < timestamp));
		}
	}

	@Override
	public void shutdown(List<NodeAgent> agents) throws RemoteException,
			NotCoordinatorException {
		while (this.coordinator == null) {
			logger.trace("unknown coordinator, calling for elections");
			bullyElection(NodeManager.getInstance().getNodeInformation(), System.currentTimeMillis());
			try {
				Thread.sleep(WAIT_FOR_BULLY_OK);
			} catch (InterruptedException e) {
				// TODO Auto-generated catch block
				e.printStackTrace();
			}
		}
		
		if (!this.coordinator.equals(NodeManager.getInstance().getNodeInformation())) {
			logger.debug("I'm not the coordinator ...");
			throw new NotCoordinatorException(this.coordinator);			
		}
		synchronized (coordinator) {
			try {
				NodeManager.getInstance().getClusterAdministration().disconnectFromGroup(agents.get(0).node());
				Balancer.balance(coordinator, agents);
				NodeManager.getInstance().unbind();
			} catch (NotBoundException e1) {
				logger.warn(e1);
			}
		}
	}

	@Override
	public void addAgentToCluster(NodeAgent agent) throws RemoteException,
			NotCoordinatorException {
		while (this.coordinator == null) {
			logger.trace("unknown coordinator, calling for elections");
			bullyElection(NodeManager.getInstance().getNodeInformation(), System.currentTimeMillis());
			try {
				Thread.sleep(WAIT_FOR_BULLY_OK);
			} catch (InterruptedException e) {
				logger.warn(e);
			}
		}
		
		if (!this.coordinator.equals(NodeManager.getInstance().getNodeInformation())) {
			logger.debug("I'm not the coordinator ...");
			throw new NotCoordinatorException(this.coordinator);			
		}
		
		synchronized (coordinator) {
			try {
				NodeManager.getInstance().getSimulation().add(agent.agent());
				Balancer.balance(coordinator);
			} catch (NotBoundException e1) {
				logger.warn(e1);
			}
		}
		
		

	}

}
